package com.sia.springcloud.heartProtect;

import com.alibaba.fastjson.JSON;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.sia.springcloud.mapper.DeviceMapper;
import com.sia.springcloud.model.Device;
import com.sia.springcloud.result.GridDataResult;
import com.sia.springcloud.service.ICollectService;
import lombok.extern.slf4j.Slf4j;
import org.quartz.DisallowConcurrentExecution;
import org.quartz.JobExecutionContext;
import org.quartz.JobExecutionException;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.scheduling.quartz.QuartzJobBean;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.locks.ReentrantLock;

/**
 * @Author: Sirius
 * @Date: 2021/12/29 21:26
 * @Description: 心跳检测保护
 **/
@Slf4j
//@Component
@DisallowConcurrentExecution   //设置不支持并发执行，因为心跳保护的原因
public class HeartbeatHandler extends QuartzJobBean {

    @Resource
    private DeviceMapper deviceMapper;

    //这里是循环依赖
    @Resource
    private ICollectService collectService;

    //其实不用也volatile也可以
    private static volatile ConcurrentHashMap<String, Map<String, InstanceInfo>> instanceInfoMapMap = new ConcurrentHashMap<>();


    //心跳保护开启标识
    private static transient boolean protectStatus = false;

    //定义锁对象
    private final ReentrantLock updateLock = new ReentrantLock();

    private final ReentrantLock getLock = new ReentrantLock();

    public boolean getProtectStatus() {
        return protectStatus;
    }

    public void setProtectStatus(boolean protect) {
        protectStatus = protect;
    }

    @Value("${heartbeatProtect.timeout:60}")
    private Integer timeout;


    /**
     * 执行心跳检测任务
     *
     * @param context
     * @throws JobExecutionException
     */
    @Override
    protected void executeInternal(JobExecutionContext context) throws JobExecutionException {
        log.info("InstanceInfoMapMap:"+instanceInfoMapMap.toString());
        if (instanceInfoMapMap.size() == 0) {
            log.info("超时处理----已启动的协议微服务个数：0");
            return;
        }

        //超时处理，如果超过20秒就删除
        int size = 0;
        long currentTime = System.currentTimeMillis();
        for (Map.Entry<String, Map<String, InstanceInfo>> mapMapEntry : instanceInfoMapMap.entrySet()) {
            Map<String, InstanceInfo> instanceInfoMap = mapMapEntry.getValue();

            for (Map.Entry<String, InstanceInfo> mapEntry : instanceInfoMap.entrySet()) {
                InstanceInfo instanceInfo = mapEntry.getValue();
                // 超时时间默认是60秒
                if (currentTime - instanceInfo.getTime() > timeout * 1000) {
                    instanceInfoMap.remove(mapEntry.getKey());
                }
            }
            log.info(mapMapEntry.getKey()+":"+instanceInfoMap.size());

            size = size + instanceInfoMap.size();
        }
        log.info("超时处理-----已启动的协议微服务个数：{}", size);


        //设置开启之后才会进行心跳保护
        if (protectStatus) {
            heartBeatDetectionProtection();
        }
    }


    /**
     * 心跳保护主程序
     * <p>
     * 首先进行超时处理
     * <p>
     * 然后寻找开启数采但是其微服务停止的设备，为其重启
     * <p>
     * 心跳检测：通过每隔一段时间向注册在Eureka上的所有协议微服务发送一个心跳检测信号，然后协议微服务返回自己的信息
     */
    private void heartBeatDetectionProtection() {

        //如果没有正在数采的设备直接返回
        QueryWrapper<Device> deviceQueryWrapper = new QueryWrapper<>();
        deviceQueryWrapper.eq("collect", 1);
        List<Device> deviceList = deviceMapper.selectList(deviceQueryWrapper);

        if (deviceList == null || deviceList.isEmpty()) {
            log.info("心跳保护-----正在数采的设备的个数：0");
            return;
        }

        //寻找开启数采但是所在微服务停止的设备，然后为其重新开启数采
        log.info("心跳保护-----正在数采的设备的个数：{}", deviceList.size());
        for (Device device : deviceList) {
            String protocolName = device.getProtocolName();
            String microservice = device.getMicroservice();
            Integer deviceId = device.getId();

            //如果不包含这个，说明没有对应的协议微服务启动
            if (!instanceInfoMapMap.containsKey(protocolName)) {
                continue;
            }

            Map<String, InstanceInfo> instanceInfoMap = instanceInfoMapMap.get(protocolName);

            //说明设备对应的协议微服务已经停止了，需要进行重启
            if (!instanceInfoMap.containsKey(microservice)) {
                log.info("心跳保护-----设备 {} 的数采微服务已经停止，即将为其重启", device.getId());


                //避免一个微服务开启任务太密集，出现并发问题
                try {
                    Thread.sleep(250);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }

                //首先对信息进行充值
                device.setCollect(0);
                device.setMicroservice("nothing");
                deviceMapper.updateById(device);

                //然后对设备进行重启
                collectService.startCollect(deviceId);
                log.info("设备 {} 进行重启", deviceId);
            }

        }
    }


    /**
     * 更新集群中微服务的信息
     * <p>
     * 首先是更新传来的实例的信息，然后是按照优先级进行排序
     * <p>
     * 同时多个微服务更新信息会造成线程同步问题
     *
     * @param newInstanceInfo
     */
    public void updateProtocolInfo(InstanceInfo newInstanceInfo) {
        log.info("更新微服务信息：{}",newInstanceInfo.toString());
        updateLock.lock();
        String protocolName = newInstanceInfo.getProtocolName();
        String heartbeatId = newInstanceInfo.getHeartbeatId();
        try {
            if (!instanceInfoMapMap.containsKey(protocolName)) {
                Map<String, InstanceInfo> newInstanceInfoMap = new HashMap<>();
                newInstanceInfoMap.put(heartbeatId, newInstanceInfo);
                instanceInfoMapMap.put(protocolName, newInstanceInfoMap);
            } else {
                //如果
                Map<String, InstanceInfo> instanceInfoMap = instanceInfoMapMap.get(protocolName);
                if(instanceInfoMap.containsKey(heartbeatId)){
                    instanceInfoMap.replace(heartbeatId, newInstanceInfo);
                }else{
                    instanceInfoMap.put(heartbeatId,newInstanceInfo);
                }
            }

            log.info("心跳检测----正常更新协议信息");
        } finally {
            updateLock.unlock();
        }


    }


    /**
     * 根据protocolName返回一个InstanceInfo（存在优先级）
     *
     * @param protocolName
     * @return 如果有则返回一个InstanceInfo，否则返回null
     */
    public InstanceInfo getInstanceInfoByProtocolName(String protocolName) {


        getLock.lock();
        try {
            InstanceInfo priorityInstanceInfo = new InstanceInfo();
            priorityInstanceInfo.setDeviceCount(Integer.MAX_VALUE);

            if (!instanceInfoMapMap.containsKey(protocolName)) {
                return null;

            } else {

                Map<String, InstanceInfo> instanceInfoMap = instanceInfoMapMap.get(protocolName);
                if (instanceInfoMap.size() == 0) {
                    return null;
                } else {
                    for (Map.Entry<String, InstanceInfo> mapEntry : instanceInfoMap.entrySet()) {
                        InstanceInfo instanceInfo = mapEntry.getValue();
                        if (instanceInfo.getDeviceCount() < priorityInstanceInfo.getDeviceCount()) {
                            priorityInstanceInfo = instanceInfo;
                        }
                    }

                    //可以考虑在这里调用updateInstanceInfo方法，对该priorityInstanceInfo进行更新
                    //其实在这里进行更新会更好，因为如果这个priorityInstanceInfo对应的数采微服务已经停止
                    //那么更新它的deviceCount反而能够避免所有设备都去这个坏的上面
                    //或许，如果startCollect方法返回失败，应该继续获取一个instanceInfo去继续尝试
                    priorityInstanceInfo.setDeviceCount(priorityInstanceInfo.getDeviceCount() + 1);
                    updateProtocolInfo(priorityInstanceInfo);

                    log.info("返回的可用的协议微服务：{}", priorityInstanceInfo.toString());

                    return priorityInstanceInfo;
                }
            }

        } finally {
            getLock.unlock();
        }
    }


    /**
     * 根据设备的microservice寻找对应的微服务实例信息
     * <p>
     * stopCollect时时调用
     *
     * @param microservice
     * @return 如果存在返回InstanceInfo，否则返回null
     */
    public InstanceInfo getInstanceInfoByMicroservice(String protocolName, String microservice) {
        if (!instanceInfoMapMap.containsKey(protocolName)) {
            return null;
        } else {
            Map<String, InstanceInfo> instanceInfoMap = instanceInfoMapMap.get(protocolName);
            if (!instanceInfoMap.containsKey(microservice)) {
                return null;
            } else {
                return instanceInfoMap.get(microservice);
            }
        }
    }


    /**
     * 获取正在运行的协议列表
     *
     * @return
     */
    public String queryProtocolList(){
        ArrayList<String> protocolList = new ArrayList<>();
        for(Map.Entry<String,Map<String, InstanceInfo>> stringMapEntry:instanceInfoMapMap.entrySet()){
            if(stringMapEntry.getValue()!=null&&stringMapEntry.getValue().size()>0){
                protocolList.add(stringMapEntry.getKey());
            }
        }
        GridDataResult<String> gridDataResult = new GridDataResult<>();
        gridDataResult.setTotal(protocolList.size());
        gridDataResult.setRoot(protocolList);
        return JSON.toJSONString(gridDataResult);
    }

}
